package com.ultrapower.request;

import java.util.List;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

import com.sematext.hbase.wd.RowKeyDistributorByHashPrefix;
import com.ultrapower.bean.DotBean;
import com.ultrapower.mapper.TransactionIdMapper;
import com.ultrapower.util.CommonUtils;
import com.ultrapower.util.HBaseTables;
import com.ultrapower.util.RowKeyUtils;
/**
 * 
 * @author wangy
 *
 */
public class ApplicationTraceIndexTable extends BaseRequest {
	private int scanCacheSize = 256;
	RowKeyDistributorByHashPrefix applicationTraceIndexDistributor = new RowKeyDistributorByHashPrefix(new RowKeyDistributorByHashPrefix.OneByteSimpleHash(32));

	public static void main(String[] args) {
		BaseRequest table = new ApplicationTraceIndexTable();
		table.execute();
	}

	@Override
	void execute() {
		TableName applicationTraceIndexTableName = this.getTableName(HBaseTables.APPLICATION_TRACE_INDEX_STR);
		this.getHbaseTemplate().find(applicationTraceIndexTableName, createScan(), applicationTraceIndexDistributor,
				(ResultScanner scanner) -> {

					TransactionIdMapper map = new TransactionIdMapper();
					scanner.forEach(r -> {
						try {
							List<DotBean> transIds = map.mapRow(r, 0);
							if (transIds != null) {
								transIds.forEach(t -> {
									System.out.println(t);
								});
							}
						} catch (Exception e) {
							e.printStackTrace();
						}
					});
					return null;
				});
	}

	@Override
	protected Scan createScan() {
		Scan scan = new Scan();
		scan.setCaching(this.scanCacheSize);

		byte[] bApplicationName = Bytes.toBytes(applicationName);
		byte[] traceIndexStartKey = RowKeyUtils.concatFixedByteAndLong(bApplicationName, HBaseTables.AGENT_NAME_MAX_LEN,
				CommonUtils.reverseTimeMillis(from));
		byte[] traceIndexEndKey = RowKeyUtils.concatFixedByteAndLong(bApplicationName, HBaseTables.AGENT_NAME_MAX_LEN,
				CommonUtils.reverseTimeMillis(to));

		scan.setReversed(true);
		scan.setStartRow(traceIndexStartKey);
		scan.setStopRow(traceIndexEndKey);

		scan.addFamily(HBaseTables.APPLICATION_TRACE_INDEX_CF_TRACE);
		scan.setId("ApplicationTraceIndexScan");

		return scan;
	}

}
